package com.peak.code;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

/**
 * @author likai
 * @date 2021/10/31 3:38 PM
 * @email likai9376@163.com
 * @desc
 */
public class Demo02 {

    public static void main(String[] args) {

//        Flux<Integer> range = Flux.range(1, 10);
//
//        SampleSubscribe sampleSubscribe = new SampleSubscribe();
//        range.subscribe(sampleSubscribe);


        Flux<Integer> r1 = Flux.range(5, 10);
        r1.doOnRequest(r-> System.out.println(r))
                .subscribe(new SampleSubscribe());

    }




    public static class SampleSubscribe extends BaseSubscriber{

        @Override
        protected Subscription upstream() {
            return super.upstream();
        }

        @Override
        public boolean isDisposed() {
            return super.isDisposed();
        }

        @Override
        public void dispose() {
            super.dispose();
        }

        //建立链接的时候，request应该至少调用一次
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            //System.out.println("hookOnSubscribe");
            request(100);
        }

        @Override
        protected void hookOnNext(Object value) {
            //request(13);
//            if(Integer.valueOf(value.toString()).intValue()<=5){
//                System.out.println("接受数据:"+value);
//                request(1);
//            }else {
//                System.out.println("接受数据:"+value);
//            }

            System.out.println("Cancelling after having received " + value);
            cancel();
        }

        @Override
        protected void hookOnComplete() {
            System.out.println("hookOnComplete");
        }

        @Override
        protected void hookOnError(Throwable throwable) {
            System.out.println("hookOnError");
        }

        @Override
        protected void hookOnCancel() {
            System.out.println("hookOnCancel");
        }

        @Override
        protected void hookFinally(SignalType type) {
            System.out.println("hookFinally");
        }
    }
}
